package com.atguigu.bigdata.spark.core.seq

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayOps

object Spark_req_hotTop10 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc = new SparkContext(sparkConf)

    //TODO TOP10热门商品

    //1.获取数据
    val dataRDD: RDD[String] = sc.textFile("data/user_visit_action.txt")


    //2.统计不同维度数据[详情点击，下单点击，支付点击 ]的数量

    //2.1品类 点击数量
    //过滤详情点击的数据
    val clickData: RDD[String] = dataRDD.filter(
      line => {
        val datas: Array[String] = line.split("_")
        datas(6) != "-1"
      }
    )
    val clickCount: RDD[(String, Int)] = clickData.map(
      line => {
        val datas: Array[String] = line.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_+_)


    //2.2品类 下单点击
    val orderData: RDD[String] = dataRDD.filter(
      line => {
        val datas: Array[String] = line.split("_")
        datas(8) != "null"
      }
    )
    val orderCount: RDD[(String, Int)] = orderData.flatMap(
      line => {
        val datas: Array[String] = line.split("_")
        val ids = datas(8).split(",")
        ids.map((_,1))

      }
    ).reduceByKey(_ + _)




    //2.3品类 支付点击
    val payData: RDD[String] = dataRDD.filter(
      line => {
        val datas: Array[String] = line.split("_")
        datas(10) != "null"
      }
    )
    val payCount: RDD[(String, Int)] = payData.flatMap(
      line => {
        val datas: ArrayOps.ofRef[String] = line.split("_")
        val ids: ArrayOps.ofRef[String] = datas(10).split(",")
        ids.map((_, 1))
      }
    ).reduceByKey(_ + _)
    payCount


    //3按照点击，下单，支付，各个排序。
/*    val sortRDD: RDD[(String, Int)] = clickCount.sortBy(_._2,false)*/
    //为了最好的比较，不使用rdd进行比较，使用元组。
 val rdd: RDD[(String, ((Int, Int), Int))] = clickCount.join(orderCount).join(payCount)
    val mapRDD: RDD[(String, (Int, Int, Int))] = rdd.mapValues {
      case ((click, order), pay) => {
        (click, order, pay)
      }
    }
    mapRDD.sortBy(_._2,false).take(10).foreach(println)




    sc.stop()
  }

}
